//package com.allwees.bs.task.order.flow;
//
//import cn.hutool.core.collection.ConcurrentHashSet;
//import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
//import com.baomidou.mybatisplus.extension.conditions.query.LambdaQueryChainWrapper;
//import com.baomidou.mybatisplus.extension.conditions.update.LambdaUpdateChainWrapper;
//import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
//import com.allwees.bs.module.oms.dao.OrderFlowRecord;
//import com.allwees.bs.module.oms.service.OrderFlowService;
//import com.xxl.job.core.biz.model.ReturnT;
//import com.xxl.job.core.handler.annotation.XxlJob;
//import lombok.Data;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.lang3.math.NumberUtils;
//import org.springframework.stereotype.Service;
//import org.springframework.util.CollectionUtils;
//
//import javax.annotation.Resource;
//import java.util.List;
//import java.util.Set;
//import java.util.concurrent.CompletableFuture;
//import java.util.concurrent.CopyOnWriteArrayList;
//import java.util.concurrent.ExecutorService;
//import java.util.concurrent.Semaphore;
//import java.util.stream.Collectors;
//
///**
// * o_order_flow由于之前逻辑错误，导致在定时任务中大量插入了垃圾数据(1亿2千万)，
// * 这个手动的task为删除这部分垃圾数据
// *
// */
//@Slf4j
//@Service
//public class DelRedundantOrderFlowRowsTask {
//
//    /**
//     * 根据规则捞取o_order_flow中的数据
//     */
//    @Resource
//    private OrderFlowService orderFlowService;
//
//    /**
//     * 执行删除任务的线程池，线程开太多mysql会扛不住
//     * 这里是单纯密集删除任务，让corePoolSize和maxPoolSize保持一致
//     */
//    @Resource
//    private ExecutorService delThreadPool;
//
//    /**
//     * 过滤阈值:当同一orderUuid的记录数大于此阈值时，认为是垃圾数据并处理
//     */
//    private static int ROWS_OF_SAME_ORDER;
//
//    /**
//     * 当前task是否处于运行状态，这里只简单在内存中做一个标识，分布式环境下应使用其他方式防止并发
//     */
//    private boolean running;
//
//    @XxlJob("delRedundantOrderFlowRows")
//    public ReturnT<String> run(String s) {
//        //xxl-job阈值参数，未传默认100，即抓取orderUuid的记录数大于100的不正常记录
//        ROWS_OF_SAME_ORDER = NumberUtils.toInt(s, 100);
//        new Thread(this::doTask).start();
//        return ReturnT.SUCCESS;
//    }
//
//    private void doTask() {
//        if (running) {
//            log.info("---------------> Task is already running...");
//            return;
//        }
//        running = true;
//        log.info(">>>>>>>>>>>>>>>>>>>> Started,threshold:{} <<<<<<<<<<<<<<<<<<<", ROWS_OF_SAME_ORDER);
//        //初始化任务上下文
//        TaskContext taskContext = initTaskContext();
//        //删除数据最大线程数，保持和线程池核心线程数一样大小，使其跑满
//        Semaphore permits = new Semaphore(ThreadPoolConfig.CONCURRENCY);
//        while (true) {
//            CurrentRedundant currentRedundant = getCurrentRedundant(taskContext,permits);
//            //由于getCurrentRedundant中可能会重新出发填充uuids操作，这里要判断当前任务是否结束
//            if (taskContext.isCompleted()) {
//                break;
//            }
//            //ForkJoinPool.commonPool()线程池中大小有限制，即使增大permits数量也达不到增加并发的目的
//            //所以执行任务时需要使用自定义的线程池
//            CompletableFuture
//                    .runAsync(() -> setMaxId(currentRedundant))
//                    .thenRun(() -> setMinId(currentRedundant))
//                    .thenRunAsync(delRedundant(currentRedundant),delThreadPool);
//        }
//        log.info(">>>>>>>>>>>>>>>>>>>> Completed,threshold:{} <<<<<<<<<<<<<<<<<<<", ROWS_OF_SAME_ORDER);
//        //标记任务结束，以便从xxl-job传递不同的阈值参数进行下一次任务
//        running = false;
//    }
//
//    private TaskContext initTaskContext(){
//        TaskContext taskContext = new TaskContext();
//        fillRedundantUuids(taskContext);
//        return taskContext;
//    }
//
//    /**
//     * 获取一个CurrentRedundant对象，并消耗一个semaphore
//     * @param taskContext
//     * @param permits
//     * @return
//     */
//    private CurrentRedundant getCurrentRedundant(TaskContext taskContext,Semaphore permits){
//        CurrentRedundant currentRedundant = new CurrentRedundant(taskContext, permits);
//        currentRedundant.acquirePermit();
//        return currentRedundant;
//    }
//
//    /**
//     * 分批循环删除该orderUuid下最大id和最小id之间的脏数据
//     * @param currentRedundant
//     * @return
//     */
//    private Runnable delRedundant(CurrentRedundant currentRedundant) {
//        return () -> {
//            while (!currentRedundant.isCompleted()) {
//                doDelRedundant(currentRedundant);
//                //从新获取最小id，以增加检索效率
//                setMinId(currentRedundant);
//            }
//            log.info("===> Removed,orderUuid:{},minId:{},maxId:{}", currentRedundant.getUuid(), currentRedundant.getMinId(), currentRedundant.getMaxId());
//            currentRedundant.releasePermit();
//        };
//    }
//
//    /**
//     * 删除该orderUuid下最大id和最小id之间的脏数据
//     * @param currentRedundant
//     */
//    private void doDelRedundant(CurrentRedundant currentRedundant) {
//        LambdaUpdateChainWrapper<OrderFlowRecord> lambdaUpdate = orderFlowService.lambdaUpdate();
//        lambdaUpdate.ge(OrderFlowRecord::getId, currentRedundant.getMinId());
//        lambdaUpdate.le(OrderFlowRecord::getId, currentRedundant.getMaxId());
//        lambdaUpdate.eq(OrderFlowRecord::getOrderUuid, currentRedundant.getUuid());
//        lambdaUpdate.last("limit 1000");
//        lambdaUpdate.remove();
//    }
//
//    /**
//     * 获取该orderUuid对应的最大的id
//     * @param currentRedundant
//     */
//    private void setMaxId(CurrentRedundant currentRedundant) {
//        LambdaQueryChainWrapper<OrderFlowRecord> lambdaQuery = orderFlowService.lambdaQuery();
//        lambdaQuery.select(OrderFlowRecord::getId, OrderFlowRecord::getAction);
//        lambdaQuery.eq(OrderFlowRecord::getOrderUuid, currentRedundant.getUuid());
//        lambdaQuery.orderByDesc(OrderFlowRecord::getId);
//        Page<OrderFlowRecord> page = lambdaQuery.page(new Page<>(1, 2, false));
//
//        List<OrderFlowRecord> records = page.getRecords();
//
//        OrderFlowRecord record = records.get(1);
//        String action = record.getAction();
//        if ("USER_CREATE".equals(action)) {
//            currentRedundant.setMaxId(record.getId());
//            log.info("===> SetMaxId,orderUuid:{},maxId:{}", currentRedundant.getUuid(), currentRedundant.getMaxId());
//            return;
//        }
//
//        if ("USER_PAYED".equals(action)) {
//            currentRedundant.setMaxId(records.get(0).getId());
//            log.info("===> SetMaxId,orderUuid:{},maxId:{}", currentRedundant.getUuid(), currentRedundant.getMaxId());
//            return;
//        }
//        currentRedundant.state = TaskState.completed;
//    }
//
//    /**
//     * 获取该orderUuid对应的最小的id
//     * @param currentRedundant
//     */
//    private void setMinId(CurrentRedundant currentRedundant) {
//        LambdaQueryChainWrapper<OrderFlowRecord> lambdaQuery = orderFlowService.lambdaQuery();
//        lambdaQuery.select(OrderFlowRecord::getId, OrderFlowRecord::getAction);
//        lambdaQuery.eq(OrderFlowRecord::getOrderUuid, currentRedundant.getUuid());
//        lambdaQuery.orderByAsc(OrderFlowRecord::getId);
//        Page<OrderFlowRecord> page = lambdaQuery.page(new Page<>(1, 3, false));
//
//        List<OrderFlowRecord> records = page.getRecords();
//
//        OrderFlowRecord record = records.get(1);
//
//        Integer minId = record.getId();
//        if (minId.equals(currentRedundant.getMaxId())) {
//            currentRedundant.state = TaskState.completed;
//            return;
//        }
//
//        if ("USER_CREATE".equals(record.getAction())) {
//            currentRedundant.setMinId(minId);
//            log.info("===> SetMinId,orderUuid:{},minId:{}", currentRedundant.getUuid(), minId);
//            return;
//        }
//
//        if ("USER_PAYED".equals(record.getAction())) {
//            if (records.size() < 3) {
//                currentRedundant.state = TaskState.completed;
//                return;
//            }
//            currentRedundant.setMinId(records.get(2).getId());
//            log.info("===> SetMinId,orderUuid:{},minId:{}", currentRedundant.getUuid(), currentRedundant.getMinId());
//            return;
//        }
//        currentRedundant.state = TaskState.completed;
//    }
//
//    /**
//     * 往TaskContext中填充新的待处理的orderUuid
//     * 当task临近结尾时，uuidsInDeleting和uuids列表中的元素会完全一样，
//     * 这里会有一段时间无效填充，待删除线程删除完毕，这里就会查不出来，也就会把任务标志为TaskState.completed
//     * 这里不做优化
//     * @param taskContext
//     */
//    private void fillRedundantUuids(TaskContext taskContext) {
//        QueryWrapper<OrderFlowRecord> qw = new QueryWrapper<>();
//        qw.select("order_uuid ,count(*) num");
//        qw.groupBy("order_uuid");
//        qw.having("num > " + ROWS_OF_SAME_ORDER);
//
//        Page<OrderFlowRecord> page = orderFlowService.page(new Page<>(1, 100, false), qw);
//
//        List<OrderFlowRecord> records = page.getRecords();
//        if (CollectionUtils.isEmpty(records)) {
//            taskContext.setTaskState(TaskState.completed);
//            return;
//        }
//        records.forEach(r -> log.info("---> orderUUid:{},redundant:{}", r.getOrderUuid(), r.getNum()));
//        taskContext.setUuids(records.stream().map(OrderFlowRecord::getOrderUuid).collect(Collectors.toCollection(CopyOnWriteArrayList::new)));
//    }
//
//    @Data
//    private static class CurrentRedundant {
//        private String uuid;
//        private TaskState state;
//        private Integer maxId;
//        private Integer minId;
//        private Semaphore permits;
//
//        private TaskContext taskContext;
//
//        CurrentRedundant(TaskContext taskContext, Semaphore permits) {
//            this.taskContext = taskContext;
//            this.permits = permits;
//            this.uuid =  taskContext.getRedundantUuid();
//            this.state = TaskState.running;
//        }
//
//        boolean isCompleted() {
//            return state == TaskState.completed;
//        }
//
//        void releasePermit() {
//            taskContext.redundantUuidDeleted(uuid);
//            permits.release();
//        }
//        void acquirePermit() {
//            permits.acquireUninterruptibly();
//        }
//    }
//
//    /**
//     * 任务上下文
//     */
//    @Data
//    private class TaskContext {
//        /**
//         * 当前任务状态
//         */
//        private volatile TaskState taskState;
//        /**
//         * 本次捞取、待处理的orderUuids
//         */
//        private volatile List<String> uuids;
//
//        /**
//         * 正在被其他线程处理的orderUuid
//         * 删除线程删除数据时，因为量比较大，需要一定的时间执行删除，
//         * 当uuids被处理空时，可能还有其他的线程还在删除数据。
//         * 当getRedundantUuid重新填充uuids时，势必会把其他线程正在处理的orderUuid重新填充，
//         * 这时新线程会拿到其他线程正在处理的orderUuid从而产生竞争(多个线程同时对一个orderUuid执行delete操作)
//         * 这里加一个正在执行删除的orderUuid的集合，保证不会产生竞争(每个线程处理一个orderUuid)
//         *
//         * 查询时最好可以把这些orderUuid排除掉，否则填充时由于mysql间隙锁的原因，效率十分低下
//         */
//        private Set<String> uuidsInDeleting = new ConcurrentHashSet<>();
//
//        TaskContext() {
//            this.taskState = TaskState.running;
//        }
//
//        boolean isCompleted() {
//            return taskState == TaskState.completed;
//        }
//
//        /**
//         * 如果taskState为已完成，这里返回的redundantUuid为null
//         * ConcurrentHashSet.contains(null)会报NE并被吃掉(无日志)，这里需要注意
//         * @return
//         */
//        String getRedundantUuid() {
//            String redundantUuid = doGet();
//            while (true){
//                if (redundantUuid == null) {
//                    break;
//                }
//                if (uuidsInDeleting.contains(redundantUuid)){
//                    redundantUuid = doGet();
//                    continue;
//                }
//                break;
//            }
//            if (redundantUuid != null) {
//                uuidsInDeleting.add(redundantUuid);
//            }
//            return redundantUuid;
//        }
//
//        /**
//         * 一条orderUuid处理完是，清除uuidsInDeleting。
//         * 因为重新填充时可能又把该uuid填充进去，为防止其他线程取到(其实取到数据库中也取不到最小id值)
//         * 所以这里也顺带把该uuid从uuids中剔除(如果有)
//         * @param uuid
//         */
//        void redundantUuidDeleted(String uuid) {
//            uuidsInDeleting.remove(uuid);
//            uuids.remove(uuid);
//        }
//
//        /**
//         * 从待处理的orderUuids中获取一个orderUuid，
//         * 如果待处理的orderUuids被处理空了，要从数据库重新获取并填充，
//         * 防止并发，这里做一个Double-Lock-Check
//         * @return
//         */
//        String doGet(){
//            if (CollectionUtils.isEmpty(uuids)) {
//                synchronized (this) {
//                    if (CollectionUtils.isEmpty(uuids)) {
//                        fillRedundantUuids(this);
//                    }
//                }
//            }
//
//            if (isCompleted()) {
//                return null;
//            }
//            //从头部获取一个元素
//            return uuids.remove(0);
//        }
//    }
//
//    private enum TaskState {
//        running, completed
//    }
//}
